#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <dirent.h>
#include <inttypes.h> 
#include <stdbool.h>
#include <errno.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include "anysocket.h"
#include "log.h"
#include "utcache.h"
#include "concurrent-connection-pool.h"
#include "networking.h"
#include "chunk-client.h"
#include "ursax.h"

struct foo_cache* ccpool = NULL;

st_netfd_t get_connection(uint32_t ip, uint16_t port)
{
	LOG_DEBUG("enter %s ip %s port %d", __func__, str_ip(ip), port);
	st_netfd_t stfd = NULL;
	uint64_t key = ccpool_make_key(ip, port);
	int ret = foo_cache_get_reference(ccpool, key, (uint64_t*)&stfd);
	if (likely(ret >= 0)) // r exist
	{
		int ref;
		int count = 0;
		while(st_is_socket_error(stfd)){
			ref = foo_cache_adjust_reference(ccpool, key, -1);
			if(ref == 0){
				// no one get this reference
				if(!st_is_socket_limit(stfd)){
no_connect_limit:
					// no time limit, get lock and goto connect
					LOG_DEBUG("get meta write permission");
					foo_cache_adjust_reference(ccpool, key, 1);
					break;
				}
				// connection limited
				if(st_is_limit_expire(stfd)){
					// connection limited, but timeout, not a effect limit
					LOG_INFO("ADDR %s:%d connection limite expired", str_ip(ip), port);
					foo_cache_change_data(ccpool, key, NULL);
					stfd = NULL;
					goto no_connect_limit;
				}
				// connection limit not timeout
				LOG_INFO("ADDR %s:%d connection limited, return NULL here", str_ip(ip), port);
				return NULL;
			}
			LOG_DEBUG("r->socket_error wait for some thread connect again addr(%s:%d) stfd %p ref is %d", str_ip(ip), port, stfd, ref);
			st_usleep(1000*1000); // 10ms -> 1s
			count++;
			if(count > 10){
				LOG_ERROR("error connect to %s:%d ref is %d and stfd get is %p count is %d", str_ip(ip), port, ref, stfd, count);
				return NULL;
			}
			ref = foo_cache_get_reference(ccpool, key, (uint64_t*)&stfd);
		}

		if(likely(!st_is_socket_error(stfd))){ // perhaps,when get the lock but r->stfd is clear by other thread, NULL get
			LOG_DEBUG("get stfd ret 1 %p", stfd);
			return stfd;
		}
		LOG_DEBUG("get_connection get stfd %p over addr(%s:%d)", stfd, str_ip(ip), port);
	} else if (ret < 0) { // stfd not exist
		foo_cache_put_reference(ccpool, key, NULL, 1);
		LOG_DEBUG("create a new ccpool record");
	}
	LOG_DEBUG("ccpool cache miss, cache size = %u", foo_cache_entry_count(ccpool));

	foo_cache_change_data(ccpool, key, NULL); // set global point to NULL
	close_connection(ip, port, stfd); // stfd will be a bad point

	st_netfd_t connection;
	MAKE_FD_ELASTIC(
			connection = ursax_connect(ip, port, true),
			"unable to create new socket",
				(foo_cache_adjust_reference(ccpool, key, -1),
				 foo_cache_change_data(ccpool, key, st_now_limit_value()),
				 (st_netfd_t)NULL)
	);

	LOG_DEBUG("set new socket stfd %p (%s:%d)", connection, str_ip(ip), port);
	foo_cache_change_data(ccpool, key, connection);
	st_alloc_rw_lock(connection, ip, port);

	LOG_DEBUG("leave %s", __func__);
	return connection;
}

void trash_connection(uint32_t ip, uint16_t port, st_netfd_t stfd)
{
	LOG_DEBUG("enter %s ip %s port %d", __func__, str_ip(ip), port);
	st_set_socket_error(stfd);

//	st_free_rw_lock(stfd); // don't close fd here!
//	st_netfd_close(stfd);

	int ref = put_connection(ip, port, stfd);
	if(ref == 0){
		uint64_t key = ccpool_make_key(ip, port);
		LOG_DEBUG("trash_connection reference is 0, %s:%d %p, close_it!!!!", str_ip(ip), port, stfd);
		foo_cache_change_data(ccpool, key, NULL);
		close_connection(ip, port, stfd);
	}
}

static void ccpool_connection_dtor(void* arg)
{
	st_netfd_t stfd= (st_netfd_t)arg;
	LOG_DEBUG("%s(%p)", __func__, stfd);
	close_connection(0, 0, stfd);
}

int cconnection_pool_init(size_t capacity)
{
	if (capacity == 0 || capacity == (size_t)-1)
		capacity = 10240;

	ccpool = foo_cache_create(capacity, &ccpool_connection_dtor, true);
	return 0;
}

int cconnection_pool_fini()
{
	return foo_cache_delete(ccpool, 0);
}
